package com.flowable.project.rocketmq;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

@Component
public class TransactionProducer {

    private String producerGroup = "trc_group";

    TransactionListener transactionListener = new TransactionListenerImpl();

    private TransactionMQProducer producer;

    //一般自定义线程池的时候需要给线程加个名称，方便以后查找问题
    ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        }
    });

    public TransactionProducer(){
        producer = new TransactionMQProducer(producerGroup);

        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);

        producer.setTransactionListener(transactionListener);

        producer.setExecutorService(executorService);

        start();
    }

    public TransactionMQProducer getProducer(){
        return this.producer;
    }

    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown(){
        this.producer.shutdown();
    }


}

class TransactionListenerImpl implements TransactionListener {

    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        System.out.println("===========executeLocalTransaction============");
        String body = new String(message.getBody());
        String key = message.getKeys();
        String transactionId = message.getTransactionId();
        System.out.println("transactionId="+transactionId+",key="+key+",body="+body);
        //执行本地事务开始 TODO

        //执行本地事务结束 TODO

        int status = Integer.parseInt(o.toString());
        if(status == 1){
            return LocalTransactionState.COMMIT_MESSAGE;
        }else if(status == 2){
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }else{
            return LocalTransactionState.UNKNOW;
        }
    }

    //当上面的方法发送状态码失败时，到达一点时间后，此方法会调用回查。
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        System.out.println("===========checkLocalTransaction=============");
        String body = new String(messageExt.getBody());
        String key = messageExt.getKeys();
        String transactionId = messageExt.getTransactionId();
        System.out.println("transactionId="+transactionId+",key="+key+",body="+body);

        //可以根据key去检查本地事务消息的状态
        //如果成功
        return LocalTransactionState.COMMIT_MESSAGE;
        //如果失败
//        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
}
